package com.collection.data;

import com.motu.vertx.module.utility.kafka.KafkaManager;
import com.motu.vertx.module.utility.toolset.Tool;
import io.vertx.core.Vertx;
import io.vertx.core.json.JsonObject;
import io.vertx.ext.web.client.WebClient;
import io.vertx.ext.web.client.WebClientOptions;
import io.vertx.kafka.client.producer.KafkaProducer;
import io.vertx.kafka.client.producer.KafkaProducerRecord;
import io.vertx.kafka.client.producer.RecordMetadata;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

public class CollectionManager {
	private static final Logger logger = LoggerFactory.getLogger(CollectionManager.class);

	public static KafkaProducer<String, String> ad_producer;
	public static KafkaProducer<String, String> bi_producer;

	public static final String KAFKA_TOPIC_COLLECTION_AD = "log_colloction_ad";
	public static final String KAFKA_TOPIC_COLLECTION_GAME = "log_colloction_bi";


//
//	public static String AD_EVENT_LOGIN = "login";//登陆事件
//	public static String AD_EVENT_RESISTER = "register";//注册事件
//	public static String AD_EVENT_PAYMENT = "payment";//付费事件
//	public static String AD_EVENT_ACTIVATE = "activate";//激活事件
//
//
//	public static String GAME_EVENT_LOGIN = "login";//登陆事件
//	public static String GAME_EVENT_RESISTER = "register";//注册事件
//	public static String GAME_EVENT_PAYMENT = "payment";//付费事件
//	public static String GAME_EVENT_ACTIVATE = "activate";//激活事件




	public static WebClient webClient;


	/**
	 * 初始化collection 服务配置- 需要在进程初始化调用
	 * @param vertx
	 * @param appConfig
	 */
	public static void init(Vertx vertx,JsonObject appConfig){
		webClient = WebClient.create(vertx, new WebClientOptions().setKeepAlive(true));

		// 数据中台 游戏行为事件 kafka地址
		if (appConfig.containsKey("bi-collection-kafka-address")){
			Properties bi_config = new Properties();
			bi_config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, appConfig.getString("bi-collection-kafka-address"));
			bi_config.put("key.serializer", StringSerializer.class);
			bi_config.put("value.serializer", StringSerializer.class);
			bi_config.put("acks", "1");
			// 优化kafka生产者配置
			KafkaManager.upgradeProducerConfig(bi_config, appConfig);
			// 创建一个Kafka Producer
			bi_producer = KafkaProducer.create(vertx, bi_config);
		}

		//  广告系统 游戏行为事件 kafka地址
		if (appConfig.containsKey("ad-collection-kafka-address")){
			Properties ad_config = new Properties();
			ad_config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, appConfig.getString("ad-collection-kafka-address"));
			ad_config.put("key.serializer", StringSerializer.class);
			ad_config.put("value.serializer", StringSerializer.class);
			ad_config.put("acks", "1");
			// 优化kafka生产者配置
			KafkaManager.upgradeProducerConfig(ad_config, appConfig);
			// 创建一个Kafka Producer
			ad_producer = KafkaProducer.create(vertx, ad_config);
		}


	}

	/***
	 * 发送魔兔广告系统(热云)事件数据
	 * @param info
	 */
	public static void addAdLog(String app_key,CollectionRequestInfo info, JsonObject chargeInfo){
		String eventName = info.getLogType();
		JsonObject jo = new JsonObject();

		String os = "other";
		String ryos = info.getRyos();

		if(ryos.equals("iOS")) {
			os = "ios";
		} else if (ryos.equals("Android")) {
			os = "android";
		}
		jo.put("app_key", app_key);
		jo.put("account_id", info.getAccountId());
		jo.put("device_id", info.getDeviceId());
		jo.put("device_type", info.getDeviceType());
		jo.put("idfa", info.getIdfa());
		jo.put("idfv", info.getIdfv());
		jo.put("caid", info.getCaid());
		jo.put("last_caid", info.getLastCaid());
		jo.put("imei", info.getImei());
		jo.put("oaid", info.getOaid());
		jo.put("android_id", info.getAndroidid());
		jo.put("ua", info.getUa());
		jo.put("os", os);
		jo.put("ip", info.getIp());
		jo.put("event_name", eventName);
		if (eventName.equals(CollectionRequestInfo.TYPE_CHARGE)) {
			jo.put("pay_amount", chargeInfo.getInteger("pay_amount"));
			jo.put("order_id", chargeInfo.getString("order_id"));
		}

		logger.info("####  CollectionManager addAdLog jo:"+jo.toString());
		pushToQueue(ad_producer,KAFKA_TOPIC_COLLECTION_AD,jo);

		//TODO 测试阶段先改成直发HTTP请求
//			webClient.postAbs("https://motuclctbeta.m2time.net/receive/" + eventName ).sendJsonObject(jo, ar -> {
//				if (ar.failed()) {
//					logger.error("####  MotuCollectionManager addCollectionLog error:"+ar.cause().getMessage());
//				}
//			});
	}

	/***
	 * 发送魔兔报表系统(数数)事件数据
	 * @param info
	 */
	public static void addBiLog(int game_id,CollectionRequestInfo info,JsonObject chargeInfo){
		String eventName = info.getLogType();
		JsonObject jo = new JsonObject();
		String os = "other";
		String ryos = info.getRyos();

		if(ryos.equals("iOS")) {
			os = "ios";
		} else if (ryos.equals("Android")) {
			os = "android";
		}
		Long curTime = System.currentTimeMillis();
		String uuid = Tool.getUUID();
		jo.put("game_id", game_id);
		jo.put("event_name", eventName);
		jo.put("time", curTime);
		jo.put("os", os);
		jo.put("ip", info.getIp());
		jo.put("device_id", info.getDeviceIdBi());
		jo.put("account_id", info.getAccountId());
		jo.put("uuid", uuid);
		jo.put("user_id", String.valueOf(info.getUserId()));


		JsonObject properties = new JsonObject();
		if (eventName.equals(CollectionRequestInfo.TYPE_LOGIN)) {
			properties.put("platform", info.getPlatform());
			properties.put("channel", info.getChannel());
			properties.put("server_id", info.getServerId());

		} else if (eventName.equals(CollectionRequestInfo.TYPE_REGISTER)) {
			properties.put("platform", info.getPlatform());
			properties.put("channel", info.getChannel());
			properties.put("server_id", info.getServerId());

		} else if (eventName.equals(CollectionRequestInfo.TYPE_CHARGE)) {
			jo.put("uuid", chargeInfo.getString("order_id"));
			properties.put("platform", info.getPlatform());
			properties.put("channel", info.getChannel());
			properties.put("server_id", info.getServerId());
			properties.put("rmb_amount",chargeInfo.getLong("rmb_amount"));
			properties.put("usd_amount", chargeInfo.getLong("usd_amount"));
			properties.put("order_id", chargeInfo.getString("order_id"));
			properties.put("gift_id", chargeInfo.getInteger("gift_id"));
			properties.put("gift_type", chargeInfo.getInteger("gift_type"));
			properties.put("pay_type", chargeInfo.getString("pay_type"));

		}
		jo.put("properties", properties);


		logger.info("####  CollectionManager addBiLog jo:"+jo.toString());
		pushToQueue(bi_producer,KAFKA_TOPIC_COLLECTION_GAME,jo);

		//TODO 测试阶段先改成直发HTTP请求
//			webClient.postAbs("https://motuclctbeta.m2time.net/receive/" + eventName ).sendJsonObject(jo, ar -> {
//				if (ar.failed()) {
//					logger.error("####  MotuCollectionManager addCollectionLog error:"+ar.cause().getMessage());
//				}
//			});
	}


	// 日志发送到消息队列
	public static void pushToQueue(KafkaProducer<String, String> producer,String topicName, JsonObject obj) {
		if (producer == null) {
			logger.error("CollectionManager pushDBQueue fail cause = producer is null!! topicName:" + topicName);
			return ;
		}
		KafkaProducerRecord<String, String> record = KafkaProducerRecord.create(topicName, obj.toString());
		producer.send(record, done -> {
			if (done.failed()) {
				logger.error("CollectionManager topicName:"+topicName+ "obj : " + obj +" pushDBQueue fail cause = " + Tool.getException(done.cause()));
			} else {
				RecordMetadata recordMetadata = done.result();
			}
		});
	}

	
}
